package com.marveal.rabbit.common.serializer.impl;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.marveal.rabbit.common.serializer.Serializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.IOException;
import java.util.Objects;

/**
 * Kryo 序列化器
 */
public class KryoSerializer implements Serializer {

    private static final Logger LOGGER = LoggerFactory.getLogger(KryoSerializer.class);

    private final Class<?> javaType;

    private static final ThreadLocal<Kryo> kryos = ThreadLocal.withInitial(() -> {
        Kryo kryo = new Kryo();
        // 支持对象循环引用（否则会栈溢出）,默认为true
        kryo.setReferences(true);
        // 不强制要求注册类, 默认为false
        // 注册行为无法保证多个 JVM 内同一个类的注册编号相同；而且业务系统中大量的 Class 也难以一一注册
        kryo.setRegistrationRequired(false);
        return kryo;
    });

    private KryoSerializer(Class<?> clazz) {
        this.javaType = clazz;
    }

    public static KryoSerializer getKryoSerializer(Class<?> clazz) {
        return new KryoSerializer(clazz);
    }

    @Override
    public byte[] serializeRaw(Object data) {
        Kryo kryo = kryos.get();
        try(ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
            Output output = new Output(baos);
            kryo.writeObjectOrNull(output, data, javaType);
            return output.getBuffer();
        } catch (Exception e) {
            LOGGER.error("#KryoSerializer.serializeRaw# serializer is error, data: {}", data);
        }
        return null;
    }

    @Override
    public String serialize(Object data) {
        byte[] bytes = serializeRaw(data);
        return new String(bytes);
    }

    @Override
    public <T> T deserializer(String content) {
        if (Objects.isNull(content)) {
            throw new RuntimeException("#KryoSerializer.deserializer# content is null");
        }
        byte[] bytes = content.getBytes();
        return deserializer(bytes);
    }

    @Override
    public <T> T deserializer(byte[] content) {
        Kryo kryo = kryos.get();
        Input input = null;
        try(ByteArrayInputStream bais = new ByteArrayInputStream(content)) {
            input = new Input(bais);
            return (T) kryo.readObjectOrNull(input, javaType);
        } catch (Exception e) {
            LOGGER.error("#KryoSerializer.deserializer# deserializer is error, content: {}", content);
        } finally {
            close(input);
        }
        return null;
    }

    private static void close(Closeable... closeables) {
        for (Closeable closeable : closeables) {
            if (closeable != null) {
                try {
                    closeable.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
